home *** CD-ROM | disk | FTP | other *** search
/ EnigmA Amiga Run 1997 February / EnigmA AMIGA RUN 15 (1997)(G.R. Edizioni)(IT)[!][issue 1997-02][PLANET CD V].iso / progs / sviluppo / python-1.4 / lib / xdrlib.py < prev    next >
Text File  |  1996-11-24  |  8KB  |  321 lines
  1. """Implements (a subset of) Sun XDR -- eXternal Data Representation.
  2.  
  3. See: RFC 1014
  4.  
  5. This module will conditionally use the _xdrmodule.so module to get
  6. support for those representations we can't do much with from Python.
  7.  
  8. """
  9.  
  10. import struct
  11. from types import LongType
  12.  
  13. # use C layer XDR libraries for some data types if available
  14. try:
  15.     import _xdr
  16. except ImportError:
  17.     _xdr = None
  18.  
  19. # this test is done to see if machine representation is the same as
  20. # network representation.  if so, we can use module struct for packing
  21. # some data types
  22. _USE_MACHINE_REP = (struct.pack('l', 1) == '\0\0\0\1')
  23.  
  24. # exceptions
  25. class Error:
  26.     """Exception class for this module. Use:
  27.  
  28.     except xdrlib.Error, var:
  29.         # var has the Error instance for the exception
  30.  
  31.     Public ivars:
  32.         msg -- contains the message
  33.  
  34.     """
  35.     def __init__(self, msg):
  36.     self.msg = msg
  37.     def __repr__(self):
  38.     return repr(self.msg)
  39.     def __str__(self):
  40.     return str(self.msg)
  41.  
  42.  
  43. class ConversionError(Error):
  44.     pass
  45.  
  46.  
  47.  
  48. class Packer:
  49.     """Pack various data representations into a buffer."""
  50.  
  51.     def __init__(self):
  52.     self.reset()
  53.  
  54.     def reset(self):
  55.     self.__buf = ''
  56.  
  57.     def get_buffer(self):
  58.     return self.__buf
  59.     # backwards compatibility
  60.     get_buf = get_buffer
  61.  
  62.     def pack_uint(self, x):
  63.     self.__buf = self.__buf + \
  64.              (chr(int(x>>24 & 0xff)) + chr(int(x>>16 & 0xff)) + \
  65.               chr(int(x>>8 & 0xff)) + chr(int(x & 0xff)))
  66.     if _USE_MACHINE_REP:
  67.     def pack_uint(self, x):
  68.         if type(x) == LongType:
  69.         x = int((x + 0x80000000L) % 0x100000000L - 0x80000000L)
  70.         self.__buf = self.__buf + struct.pack('l', x)
  71.  
  72.     pack_int = pack_uint
  73.     pack_enum = pack_int
  74.  
  75.     def pack_bool(self, x):
  76.     if x: self.__buf = self.__buf + '\0\0\0\1'
  77.     else: self.__buf = self.__buf + '\0\0\0\0'
  78.  
  79.     def pack_uhyper(self, x):
  80.     self.pack_uint(int(x>>32 & 0xffffffff))
  81.     self.pack_uint(int(x & 0xffffffff))
  82.  
  83.     pack_hyper = pack_uhyper
  84.  
  85.     def pack_float(self, x):
  86.     raise ConversionError('Not supported')
  87.     def pack_double(self, x):
  88.     raise ConversionError('Not supported')
  89.     # get these from the C layer if available
  90.     if _xdr:
  91.     def pack_float(self, x):
  92.         try: self.__buf = self.__buf + _xdr.pack_float(x)
  93.         except _xdr.error, msg:
  94.         raise ConversionError(msg)
  95.     def pack_double(self, x):
  96.         try: self.__buf = self.__buf + _xdr.pack_double(x)
  97.         except _xdr.error, msg:
  98.         raise ConversionError(msg)
  99.  
  100.     def pack_fstring(self, n, s):
  101.     if n < 0:
  102.         raise ValueError, 'fstring size must be nonnegative'
  103.     n = ((n+3)/4)*4
  104.     data = s[:n]
  105.     data = data + (n - len(data)) * '\0'
  106.     self.__buf = self.__buf + data
  107.  
  108.     pack_fopaque = pack_fstring
  109.  
  110.     def pack_string(self, s):
  111.     n = len(s)
  112.     self.pack_uint(n)
  113.     self.pack_fstring(n, s)
  114.  
  115.     pack_opaque = pack_string
  116.     pack_bytes = pack_string
  117.  
  118.     def pack_list(self, list, pack_item):
  119.     for item in list:
  120.         self.pack_uint(1)
  121.         pack_item(item)
  122.     self.pack_uint(0)
  123.  
  124.     def pack_farray(self, n, list, pack_item):
  125.     if len(list) <> n:
  126.         raise ValueError, 'wrong array size'
  127.     for item in list:
  128.         pack_item(item)
  129.  
  130.     def pack_array(self, list, pack_item):
  131.     n = len(list)
  132.     self.pack_uint(n)
  133.     self.pack_farray(n, list, pack_item)
  134.  
  135.  
  136.  
  137. class Unpacker:
  138.     """Unpacks various data representations from the given buffer."""
  139.  
  140.     def __init__(self, data):
  141.     self.reset(data)
  142.  
  143.     def reset(self, data):
  144.     self.__buf = data
  145.     self.__pos = 0
  146.  
  147.     def get_position(self):
  148.     return self.__pos
  149.  
  150.     def set_position(self, position):
  151.     self.__pos = position
  152.  
  153.     def done(self):
  154.     if self.__pos < len(self.__buf):
  155.         raise Error('unextracted data remains')
  156.  
  157.     def unpack_uint(self):
  158.     i = self.__pos
  159.     self.__pos = j = i+4
  160.     data = self.__buf[i:j]
  161.     if len(data) < 4:
  162.         raise EOFError
  163.     x = long(ord(data[0]))<<24 | ord(data[1])<<16 | \
  164.         ord(data[2])<<8 | ord(data[3])
  165.     # Return a Python long only if the value is not representable
  166.     # as a nonnegative Python int
  167.     if x < 0x80000000L:
  168.         x = int(x)
  169.     return x
  170.     if _USE_MACHINE_REP:
  171.     def unpack_uint(self):
  172.         i = self.__pos
  173.         self.__pos = j = i+4
  174.         data = self.__buf[i:j]
  175.         if len(data) < 4:
  176.         raise EOFError
  177.         return struct.unpack('l', data)[0]
  178.  
  179.     def unpack_int(self):
  180.     x = self.unpack_uint()
  181.     if x >= 0x80000000L:
  182.         x = x - 0x100000000L
  183.     return int(x)
  184.  
  185.     unpack_enum = unpack_int
  186.     unpack_bool = unpack_int
  187.  
  188.     def unpack_uhyper(self):
  189.     hi = self.unpack_uint()
  190.     lo = self.unpack_uint()
  191.     return long(hi)<<32 | lo
  192.  
  193.     def unpack_hyper(self):
  194.     x = self.unpack_uhyper()
  195.     if x >= 0x8000000000000000L:
  196.         x = x - 0x10000000000000000L
  197.     return x
  198.  
  199.     def unpack_float(self):
  200.     raise ConversionError('Not supported')
  201.     def unpack_double(self):
  202.     raise ConversionError('Not supported')
  203.     # get these from the C layer if available
  204.     if _xdr:
  205.     def unpack_float(self):
  206.         i = self.__pos
  207.         self.__pos = j = i+4
  208.         data = self.__buf[i:j]
  209.         if len(data) < 4:
  210.         raise EOFError
  211.         try: return _xdr.unpack_float(data)
  212.         except _xdr.error, msg:
  213.         raise ConversionError(msg)
  214.  
  215.     def unpack_double(self):
  216.         i = self.__pos
  217.         self.__pos = j = i+8
  218.         data = self.__buf[i:j]
  219.         if len(data) < 8:
  220.         raise EOFError
  221.         try: return _xdr.unpack_double(data)
  222.         except _xdr.error, msg:
  223.         raise ConversionError(msg)
  224.  
  225.     def unpack_fstring(self, n):
  226.     if n < 0:
  227.         raise ValueError, 'fstring size must be nonnegative'
  228.     i = self.__pos
  229.     j = i + (n+3)/4*4
  230.     if j > len(self.__buf):
  231.         raise EOFError
  232.     self.__pos = j
  233.     return self.__buf[i:i+n]
  234.  
  235.     unpack_fopaque = unpack_fstring
  236.  
  237.     def unpack_string(self):
  238.     n = self.unpack_uint()
  239.     return self.unpack_fstring(n)
  240.  
  241.     unpack_opaque = unpack_string
  242.     unpack_bytes = unpack_string
  243.  
  244.     def unpack_list(self, unpack_item):
  245.     list = []
  246.     while 1:
  247.         x = self.unpack_uint()
  248.         if x == 0: break
  249.         if x <> 1:
  250.         raise ConversionError('0 or 1 expected, got ' + `x`)
  251.         item = unpack_item()
  252.         list.append(item)
  253.     return list
  254.  
  255.     def unpack_farray(self, n, unpack_item):
  256.     list = []
  257.     for i in range(n):
  258.         list.append(unpack_item())
  259.     return list
  260.  
  261.     def unpack_array(self, unpack_item):
  262.     n = self.unpack_uint()
  263.     return self.unpack_farray(n, unpack_item)
  264.  
  265.  
  266. # test suite
  267. def _test():
  268.     p = Packer()
  269.     packtest = [
  270.     (p.pack_uint,    (9,)),
  271.     (p.pack_bool,    (None,)),
  272.     (p.pack_bool,    ('hello',)),
  273.     (p.pack_uhyper,  (45L,)),
  274.     (p.pack_float,   (1.9,)),
  275.     (p.pack_double,  (1.9,)),
  276.     (p.pack_string,  ('hello world',)),
  277.     (p.pack_list,    (range(5), p.pack_uint)),
  278.     (p.pack_array,   (['what', 'is', 'hapnin', 'doctor'], p.pack_string)),
  279.     ]
  280.     succeedlist = [1] * len(packtest)
  281.     count = 0
  282.     for method, args in packtest:
  283.     print 'pack test', count,
  284.     try:
  285.         apply(method, args)
  286.         print 'succeeded'
  287.     except ConversionError, var:
  288.         print 'ConversionError:', var.msg
  289.         succeedlist[count] = 0
  290.     count = count + 1
  291.     data = p.get_buffer()
  292.     # now verify
  293.     up = Unpacker(data)
  294.     unpacktest = [
  295.     (up.unpack_uint,   (), lambda x: x == 9),
  296.     (up.unpack_bool,   (), lambda x: not x),
  297.     (up.unpack_bool,   (), lambda x: x),
  298.     (up.unpack_uhyper, (), lambda x: x == 45L),
  299.     (up.unpack_float,  (), lambda x: 1.89 < x < 1.91),
  300.     (up.unpack_double, (), lambda x: 1.89 < x < 1.91),
  301.     (up.unpack_string, (), lambda x: x == 'hello world'),
  302.     (up.unpack_list,   (up.unpack_uint,), lambda x: x == range(5)),
  303.     (up.unpack_array,  (up.unpack_string,),
  304.      lambda x: x == ['what', 'is', 'hapnin', 'doctor']),
  305.     ]
  306.     count = 0
  307.     for method, args, pred in unpacktest:
  308.     print 'unpack test', count,
  309.     try:
  310.         if succeedlist[count]:
  311.         x = apply(method, args)
  312.         print pred(x) and 'succeeded' or 'failed', ':', x
  313.         else:
  314.         print 'skipping'
  315.     except ConversionError, var:
  316.         print 'ConversionError:', var.msg
  317.     count = count + 1
  318.  
  319. if __name__ == '__main__':
  320.     _test()
  321.